#include "thread_pool.h"

using namespace std;

ThreadPool::ThreadPool()
    : mMinNum(0)
    , mMaxNum(0) {
}

ThreadPool::~ThreadPool() {
}

void ThreadPool::Init(size_t minThreadNum, size_t maxThreadNum) {
    assert(0 == mMinNum && 0 == mMaxNum);
    assert(minThreadNum > 0);
    assert(maxThreadNum >= minThreadNum);

    mMinNum = minThreadNum;
    mMaxNum = maxThreadNum;

    SmartLock lock(mMutex);
    for (size_t i = 0; i < minThreadNum; ++i) {
        mThreads.push_back(new ThreadHelper(this));
    }
}

void ThreadPool::Join() {
    vector<ThreadInfo> tmp;
    {
        SmartLock lock(mMutex);
        for (vector<ThreadInfo>::iterator it = mThreads.begin(); it != mThreads.end(); ++it) {
            mTasks.Enqueue(ThreadData());
            tmp.push_back(*it);
        }
        mThreads.clear();
    }
    for (vector<ThreadInfo>::iterator it = tmp.begin(); it != tmp.end(); ++it) {
        it->mThread->Join();
        delete it->mThread;
    }
}

void ThreadPool::Execute(IThreadTask* task, void* param) {
    assert(task);

    vector<ThreadInfo> tmp;
    {
        SmartLock lock(mMutex);
        if (mTasks.Enqueue(ThreadData(task, param)) > mMinNum) {
            if (mThreads.size() >= mMinNum && mThreads.size() < mMaxNum) {
                mThreads.push_back(new ThreadHelper(this));
            }
        }

        vector<ThreadInfo>::iterator it = mThreads.begin();
        while (it != mThreads.end()) {
            if (it->mEnable) {
                ++it;
            } else {
                tmp.push_back(*it);
                it = mThreads.erase(it);
            }
        }
    }
    for (vector<ThreadInfo>::iterator it = tmp.begin(); it != tmp.end(); ++it) {
        it->mThread->Join();
        delete it->mThread;
    }
}

void ThreadPool::OnThread(const ThreadHelper& sender) {
    ThreadData data;
    while (1) {
        if (mTasks.Dequeue(data, 300000)) {
            if (data.mTask) {
                data.mTask->OnThread(data.mParam);
            } else {
                break;
            }
        } else {
            SmartLock lock(mMutex);
            size_t count = 0;
            vector<ThreadInfo>::iterator curr = mThreads.end();
            for (vector<ThreadInfo>::iterator it = mThreads.begin(); it != mThreads.end(); ++it) {
                if (it->mEnable)
                    ++count;

                if (&sender == it->mThread)
                    curr = it;
            }
            if (count > mMinNum && curr != mThreads.end()) {
                curr->mEnable = false;
                break;
            }
        }
    }
}
